package com.glodon.paas.job.server.service.impl;

import static com.glodon.paas.job.JobConstants.*;
import static com.glodon.paas.job.util.JobUtil.*;

import java.util.Enumeration;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.broker.BrokerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import com.glodon.paas.job.JobConstants;
import com.glodon.paas.job.model.Task;
import com.glodon.paas.job.model.TaskStatus;
import com.glodon.paas.job.server.dao.TaskDAO;
import com.glodon.paas.job.server.service.TaskService;
import com.glodon.paas.job.util.JobUtil;

@Service
public class JmsTaskService implements TaskService {
    @Autowired
    private TaskDAO taskDAO;
    @Value("${job.server.activemq.embedded}")
    private boolean activeMqEmbedded;
    @Value("${job.server.activemq.url}")
    private String activemqUrl;
    @Autowired
    private JmsTemplate jmsTemplate;
    private BrokerService brokerService;

    public JmsTaskService() throws Exception {
    }

    @PostConstruct
    public void init() throws Exception {
        System.setProperty("job.server.activemq.url", activemqUrl);
        if (activeMqEmbedded) {
            brokerService = createEmbeddedActiveMq(activemqUrl);
        }
    }

    @Override
    public Task addTask(final String type, final Map<?, ?> params) {
        String queueName = TASK_QUEUE_PREFIX + type;
        final Task task = Task.createTask(type, params);
        taskDAO.save(task);
        jmsTemplate.send(queueName, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                return task.toMessage(session);
            }
        });
        return task;
    }

    @Override
    public boolean cancelTask(Task task) {
        task.setStatus(TaskStatus.canceled);
        taskDAO.saveOrUpdate(task);
        String queueName = TASK_QUEUE_PREFIX + task.getTaskType();
        Message msg = jmsTemplate.receiveSelected(queueName, PROPERTY_TASK_ID + " = '" + task.getTaskId() + "'");
        task.setEndTime(JobUtil.getMessageOutTime(msg));
        if (msg != null) {
            toRecycle(msg);
        }
        return msg != null;
    }

    private void toRecycle(final Message msg) {
        jmsTemplate.send(JobConstants.RECYCLE_QUEUE, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                return copyMessage(msg, session);
            }
        });
    }

    private Message copyMessage(final Message msg, Session session) throws JMSException {
        Message recycleMsg;
        if (msg instanceof TextMessage) {
            TextMessage textMsg = (TextMessage) msg;
            TextMessage recycleTextMsg = session.createTextMessage();
            recycleTextMsg.setText(textMsg.getText());
            recycleMsg = recycleTextMsg;
        } else {
            recycleMsg = session.createMessage();
        }
        Enumeration names = msg.getPropertyNames();
        while (names.hasMoreElements()) {
            String name = (String) names.nextElement();
            recycleMsg.setObjectProperty(name, msg.getObjectProperty(name));
        }
        return recycleMsg;
    }

    @Override
    public boolean delTask(Task task) {
        return taskDAO.delete(task);
    }

    public void setActiveMqEmbedded(boolean activeMqEmbedded) {
        this.activeMqEmbedded = activeMqEmbedded;
    }

    public void setActivemqUrl(String activemqUrl) {
        this.activemqUrl = activemqUrl;
    }

    public void setTaskDAO(TaskDAO taskDAO) {
        this.taskDAO = taskDAO;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    @Override
    public Task getTask(String id) {
        return taskDAO.get(id);
    }

    @Override
    public List<Task> getAllTask() {
        return taskDAO.getAll();
    }

    @Override
    public void save(Task task) {
        taskDAO.saveOrUpdate(task);
    }

    @PreDestroy
    public void stop() throws Exception {
        if (activeMqEmbedded)
            brokerService.stop();
    }

}
